package cn.doitedu.demo.base;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class Demo1_TableApi {
    public static void main(String[] args) {

        // 构造编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射数据源（kafka中的ss-1）
        TableDescriptor tableDescriptor = TableDescriptor
                .forConnector("kafka")   // 指定连接器
                .option("topic", "ss-1")  // 指定连接器的参数
                .option("properties.bootstrap.servers", "doitedu:9092")
                .option("properties.group.id", "doit44_g2")
                .option("scan.startup.mode", "latest-offset")
                .option("value.format", "json")
                .option("value.fields-include", "EXCEPT_KEY")
                .schema(Schema.newBuilder()  // 指定表结构schema
                        .column("uid", DataTypes.BIGINT())
                        .column("event_id", DataTypes.STRING())
                        .column("properties", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
                        .column("action_time", DataTypes.BIGINT())
                        .build()
                )
                .build();

        Table sourceTable = tenv.from(tableDescriptor);

        // 按照需求，进行数据运算： 过滤出所有的来访页为 /x 的页面浏览事件
        TableResult tableResult = sourceTable
                .where(
                        $("event_id").isEqual("page_load")  // event_id = 'page_load'
                                .and($("properties").at("ref").isEqual("/x")   // properties['ref']='/x'
                                ))
                .select($("uid"), $("event_id"), $("properties"), $("action_time"))
                .execute();

        tableResult.print();

    }
}
